package com.peaksport.framework.extend.kafka;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;

/**
 * 扩展KafkaListenerAnnotationBeanPostProcessor处理KafkaListener注解
 * @author jinbin_guo  2019年5月28日 上午11:18:30
 *
 * @param <K>
 * @param <V>
 */
public class PKKafkaListenerAnnotationBeanPostProcessor<K,V> extends KafkaListenerAnnotationBeanPostProcessor<K, V> {
	
	/** 是否测试环境 */
	@Value("${peak.pkms.isTest:false}")
	private boolean isTest;
	
	/**
	 * 动态调整topic的名称,当在测试环境时自动在topic前加"Test.",这与kafak发送消息topic调整相匹配
	 */
	protected void processListener(org.springframework.kafka.config.MethodKafkaListenerEndpoint<?,?> endpoint, KafkaListener kafkaListener, Object bean, Object adminTarget, String beanName) {
		super.processListener(endpoint, kafkaListener, bean, adminTarget, beanName);
		String[] topics = kafkaListener.topics();
		//测试环境自动在topic前加"Test."
		for (int i = 0, len = topics.length; isTest && i < len; i++) 
			topics[i] = "Test." + topics[i];
		endpoint.setTopics(topics);
		
		
		if (bean instanceof PKKafkaReceiver) {
			String topic = ((PKKafkaReceiver)bean).getTopic();
			String groupId = ((PKKafkaReceiver)bean).getGroupId();
			if (isTest) topic = "Test." + topic;
			endpoint.setTopics(new String[] {topic});
			endpoint.setGroupId(groupId);
		}
	}
}
